package com.ctsi.canal.starter.client;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.ctsi.canal.starter.handler.MessageHandler;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * canal抽象客户端
 * @author ajh
 * @date 2019/10/24 15:24
 */
public abstract class AbstractCanalClient implements CanalClient {
    private Logger log = LoggerFactory.getLogger(AbstractCanalClient.class);

    protected volatile boolean flag;

    private Thread workThread;

    private CanalConnector connector;

    protected String filter = StringUtils.EMPTY;

    protected Integer batchSize = 1;

    protected Long timeout = 1L;

    protected TimeUnit unit = TimeUnit.SECONDS;

    private MessageHandler messageHandler;

    private boolean isRunning = false;

    @Override
    public void start() {
        log.info("start canal client");
        workThread = new Thread(this::process);
        workThread.setName("canal-client-thread");
        flag = true;
        workThread.start();
    }

    @Override
    public void stop() {
        log.info("stop canal client");
        flag = false;
        if (null != workThread) {
            workThread.interrupt();
        }

    }

    @Override
    @SuppressWarnings("unchecked")
    public void process() {
        while (flag) {
            try {
                if (!isRunning) {
                    connector.connect();
                    connector.subscribe(filter);
                    connector.rollback();
                    isRunning = true;
                    log.info("canal server 连接成功!");
                }

                while (flag) {
                    long batchId = 0;
                    Message message = connector.getWithoutAck(batchSize, timeout, unit);
                    batchId = message.getId();
                    log.info("获取消息 {}", batchId);
                    try {
                        if (batchId != -1 && !message.getEntries().isEmpty()) {
                            messageHandler.handleMessage(message);
                        }
                        connector.ack(batchId);
                    } catch (CanalClientException e) {
                        log.error("canal消息处理异常，{}", e.getMessage());
                        connector.rollback(batchId);
                    }
                }
            } catch (Exception e) {
                log.error("canal server 连接异常", e);
                e.printStackTrace();
                isRunning = false;
            } finally {
                connector.disconnect();
            }
        }
    }


    public void setConnector(CanalConnector connector) {
        this.connector = connector;
    }


    public void setMessageHandler(MessageHandler messageHandler) {
        this.messageHandler = messageHandler;
    }


    public CanalConnector getConnector() {
        return connector;
    }


    public MessageHandler getMessageHandler() {
        return messageHandler;
    }
}
